ScaDaMaLe Course site and book

This notebook is from databricks document: - https://docs.databricks.com/_static/notebooks/package-cells.html

As you know, we need to eventually package and deploy our models, say using sbt.

However it is nice to be in a notebook environment to prototype and build intuition and create better pipelines.

Using package cells we can be in the best of both worlds to an extent.

Package Cells

Package cells are special cells that get compiled when executed. These cells have no visibility with respect to the rest of the notebook. You may think of them as separate scala files.

This means that only class and object definitions may go inside this cell. You may not have any variable or function definitions lying around by itself. The following cell will not work.

If you wish to use custom classes and/or objects defined within notebooks reliably in Spark, and across notebook sessions, you must use package cells to define those classes.

Unless you use package cells to define classes, you may also come across obscure bugs as follows:

// We define a class
case class TestKey(id: Long, str: String)

defined class TestKey

// we use that class as a key in the group by
val rdd = sc.parallelize(Array((TestKey(1L, "abd"), "dss"), (TestKey(2L, "ggs"), "dse"), (TestKey(1L, "abd"), "qrf")))

rdd.groupByKey().collect

rdd: org.apache.spark.rdd.RDD[(TestKey, String)] = ParallelCollectionRDD[0] at parallelize at command-2971213210276613:2
res0: Array[(TestKey, Iterable[String])] = Array((TestKey(1,abd),CompactBuffer(dss)), (TestKey(1,abd),CompactBuffer(qrf)), (TestKey(2,ggs),CompactBuffer(dse)))

What went wrong above? Even though we have two elements for the key TestKey(1L, "abd"), they behaved as two different keys resulting in:

Array[(TestKey, Iterable[String])] = Array(
  (TestKey(2,ggs),CompactBuffer(dse)), 
  (TestKey(1,abd),CompactBuffer(dss)), 
  (TestKey(1,abd),CompactBuffer(qrf)))

Once we define our case class within a package cell, we will not face this issue.

package com.databricks.example

case class TestKey(id: Long, str: String)

Warning: classes defined within packages cannot be redefined without a cluster restart.
Compilation successful.

import com.databricks.example

val rdd = sc.parallelize(Array(
  (example.TestKey(1L, "abd"), "dss"), (example.TestKey(2L, "ggs"), "dse"), (example.TestKey(1L, "abd"), "qrf")))

rdd.groupByKey().collect

import com.databricks.example
rdd: org.apache.spark.rdd.RDD[(com.databricks.example.TestKey, String)] = ParallelCollectionRDD[2] at parallelize at command-2971213210276616:3
res2: Array[(com.databricks.example.TestKey, Iterable[String])] = Array((TestKey(1,abd),CompactBuffer(dss, qrf)), (TestKey(2,ggs),CompactBuffer(dse)))

As you can see above, the group by worked above, grouping two elements (dss, qrf) under TestKey(1,abd).

These cells behave as individual source files, therefore only classes and objects can be defined inside these cells.

package x.y.z

val aNumber = 5 // won't work

def functionThatWillNotWork(a: Int): Int = a + 1

The following cell is the way to go.

package x.y.z

object Utils {
  val aNumber = 5 // works!
  def functionThatWillWork(a: Int): Int = a + 1
}

Warning: classes defined within packages cannot be redefined without a cluster restart.
Compilation successful.

import x.y.z.Utils

Utils.functionThatWillWork(Utils.aNumber)

import x.y.z.Utils
res5: Int = 6

Why did we get the warning: classes defined within packages cannot be redefined without a cluster restart?

Classes that get compiled with the package cells get dynamically injected into Spark's classloader. Currently it's not possible to remove classes from Spark's classloader. Any classes that you define and compile will have precedence in the classloader, therefore once you recompile, it will not be visible to your application.

Well that kind of beats the purpose of iterative notebook development if I have to restart the cluster, right? In that case, you may just rename the package to x.y.z2 during development/fast iteration and fix it once everything works.

One thing to remember with package cells is that it has no visiblity regarding the notebook environment.

• The SparkContext will not be defined as sc.
• The SQLContext will not be defined as sqlContext.
• Did you import a package in a separate cell? Those imports will not be available in the package cell and have to be remade.
• Variables imported through %run cells will not be available.

It is really a standalone file that just looks like a cell in a notebook. This means that any function that uses anything that was defined in a separate cell, needs to take that variable as a parameter or the class needs to take it inside the constructor.

package x.y.zpackage

import org.apache.spark.SparkContext

case class IntArray(values: Array[Int])

class MyClass(sc: SparkContext) {
  def sparkSum(array: IntArray): Int = {
    sc.parallelize(array.values).reduce(_ + _)
  }
}

object MyClass {
  def sparkSum(sc: SparkContext, array: IntArray): Int = {
    sc.parallelize(array.values).reduce(_ + _)
  }
}

Warning: classes defined within packages cannot be redefined without a cluster restart.
Compilation successful.

import x.y.zpackage._

val array = IntArray(Array(1, 2, 3, 4, 5))

val myClass = new MyClass(sc)
myClass.sparkSum(array)

import x.y.zpackage._
array: x.y.zpackage.IntArray = IntArray([I@44a91426)
myClass: x.y.zpackage.MyClass = x.y.zpackage.MyClass@5d3b142
res7: Int = 15

MyClass.sparkSum(sc, array)

res8: Int = 15

Build Packages Locally

Although package cells are quite handy for quick prototyping in notebook environemnts, it is better to develop packages locally on your laptop and then upload the packaged or assembled jar file into databricks to use the classes and methods developed in the package.

ScaDaMaLe Course site and book

Introduction to Simulation

Core ideas in Monte Carlo simulation

• modular arithmetic gives pseudo-random streams that are indistiguishable from 'true' Uniformly distributed samples in integers from \({0,1,2,...,m}\)
• by diving the integer streams from above by \(m\) we get samples from \({0/m,1/m,...,(m-1)/m}\) and "pretend" this to be samples from the Uniform(0,1) RV
• we can use inverse distribution function of von Neumann's rejection sampler to convert samples from Uniform(0,1) RV to the following:
  • any other random variable
  • vector of random variables that could be dependent
  • or more generally other random structures:
    • random graphs and networks
    • random walks or (sensible perturbations of live traffic data on open street maps for hypothesis tests)
    • models of interacting paticle systems in ecology / chemcal physics, etc...

def frameIt( u:String, h:Int ) : String = {
      """<iframe 
 src=""""+ u+""""
 width="95%" height="""" + h + """"
 sandbox>
  <p>
    <a href="http://spark.apache.org/docs/latest/index.html">
      Fallback link for browsers that, unlikely, don't support frames
    </a>
  </p>
</iframe>"""
   }
displayHTML(frameIt("https://en.wikipedia.org/wiki/Uniform_distribution_(continuous)",500))

• https://en.wikipedia.org/wiki/Inversetransformsampling
• https://en.wikipedia.org/wiki/Rejection_sampling - will revisit below for Expoential RV

breeze.stats.distributions

Breeze also provides a fairly large number of probability distributions. These come with access to probability density function for either discrete or continuous distributions. Many distributions also have methods for giving the mean and the variance.

import breeze.stats.distributions._

val poi = new Poisson(3.0);

import breeze.stats.distributions._
poi: breeze.stats.distributions.Poisson = Poisson(3.0)

val s = poi.sample(5); // let's draw five samples - black-box

s: IndexedSeq[Int] = Vector(2, 4, 3, 4, 6)

Getting probabilities of the Poisson samples

s.map( x => poi.probabilityOf(x) ) // PMF

res0: IndexedSeq[Double] = Vector(0.22404180765538775, 0.16803135574154085, 0.22404180765538775, 0.16803135574154085, 0.05040940672246224)

val doublePoi = for(x <- poi) yield x.toDouble // meanAndVariance requires doubles, but Poisson samples over Ints

doublePoi: breeze.stats.distributions.Rand[Double] = MappedRand(Poisson(3.0),$Lambda$5909/1062631843@70cdbd4c)

breeze.stats.meanAndVariance(doublePoi.samples.take(1000));

res1: breeze.stats.meanAndVariance.MeanAndVariance = MeanAndVariance(3.0660000000000034,3.250894894894892,1000)

(poi.mean, poi.variance) // population mean and variance

res2: (Double, Double) = (3.0,3.0)

Exponential random Variable

Let's focus on getting our hands direty with a common random variable.

NOTE: Below, there is a possibility of confusion for the term rate in the family of exponential distributions. Breeze parameterizes the distribution with the mean, but refers to it as the rate.

val expo = new Exponential(0.5);

expo: breeze.stats.distributions.Exponential = Exponential(0.5)

expo.rate // what is the rate parameter

res3: Double = 0.5

A characteristic of exponential distributions is its half-life, but we can compute the probability a value falls between any two numbers.

expo.probability(0, math.log(2) * expo.rate)

res4: Double = 0.5

expo.probability(math.log(2) * expo.rate, 10000.0)

res5: Double = 0.5

expo.probability(0.0, 1.5)

res6: Double = 0.950212931632136

The above result means that approximately 95% of the draws from an exponential distribution fall between 0 and thrice the mean. We could have easily computed this with the cumulative distribution as well.

1 - math.exp(-3.0) // the CDF of the Exponential RV with rate parameter 3

res7: Double = 0.950212931632136

Drawing samples from Exponential RV

val samples = expo.sample(2).sorted; // built-in black box - we will roll our own shortly in Spark

samples: IndexedSeq[Double] = Vector(2.412493444974671, 3.6802759985818687)

expo.probability(samples(0), samples(1));

res8: Double = 0.007390811722912227

breeze.stats.meanAndVariance(expo.samples.take(10000)); // mean and variance of the sample

res9: breeze.stats.meanAndVariance.MeanAndVariance = MeanAndVariance(2.0109026469746554,4.073504140726905,10000)

(1 / expo.rate, 1 / (expo.rate * expo.rate)) // mean and variance of the population

res10: (Double, Double) = (2.0,4.0)

Pseudo Random Numbers in Spark

import spark.implicits._
import org.apache.spark.sql.functions._

import spark.implicits._
import org.apache.spark.sql.functions._

val df = spark.range(1000).toDF("Id") // just make a DF of 1000 row indices

df: org.apache.spark.sql.DataFrame = [Id: bigint]

df.show(5)

+---+
| Id|
+---+
|  0|
|  1|
|  2|
|  3|
|  4|
+---+
only showing top 5 rows

val dfRand = df.select($"Id", rand(seed=1234567) as "rand") // add a column of random numbers in (0,1)

dfRand: org.apache.spark.sql.DataFrame = [Id: bigint, rand: double]

dfRand.show(5) // these are first 5 of the 1000 samples from the Uniform(0,1) RV

+---+--------------------+
| Id|                rand|
+---+--------------------+
|  0|0.024042816995877625|
|  1|  0.4763832311243641|
|  2|  0.3392911406256911|
|  3|  0.6282132043405342|
|  4|  0.9665960987271114|
+---+--------------------+
only showing top 5 rows

val dfRand = df.select($"Id", rand(seed=1234567) as "rand") // add a column of random numbers in (0,1)
dfRand.show(5) // these are first 5 of the 1000 samples from the Uniform(0,1) RV

+---+--------------------+
| Id|                rand|
+---+--------------------+
|  0|0.024042816995877625|
|  1|  0.4763832311243641|
|  2|  0.3392911406256911|
|  3|  0.6282132043405342|
|  4|  0.9665960987271114|
+---+--------------------+
only showing top 5 rows

dfRand: org.apache.spark.sql.DataFrame = [Id: bigint, rand: double]

val dfRand = df.select($"Id", rand(seed=879664) as "rand") // add a column of random numbers in (0,1)
dfRand.show(5) // these are first 5 of the 1000 samples from the Uniform(0,1) RV

+---+-------------------+
| Id|               rand|
+---+-------------------+
|  0|  0.269224348263137|
|  1|0.20433965628039852|
|  2| 0.8481228617934538|
|  3| 0.6665103087537884|
|  4| 0.7712898780552342|
+---+-------------------+
only showing top 5 rows

dfRand: org.apache.spark.sql.DataFrame = [Id: bigint, rand: double]

Let's use the inverse CDF of the Exponential RV to transform these samples from the Uniform(0,1) RV into those from the Exponential RV.

val dfRand = df.select($"Id", rand(seed=1234567) as "rand") // add a column of random numbers in (0,1)
               .withColumn("one",lit(1.0))
               .withColumn("rate",lit(0.5))

dfRand: org.apache.spark.sql.DataFrame = [Id: bigint, rand: double ... 2 more fields]

dfRand.show(5) 

+---+--------------------+---+----+
| Id|                rand|one|rate|
+---+--------------------+---+----+
|  0|0.024042816995877625|1.0| 0.5|
|  1|  0.4763832311243641|1.0| 0.5|
|  2|  0.3392911406256911|1.0| 0.5|
|  3|  0.6282132043405342|1.0| 0.5|
|  4|  0.9665960987271114|1.0| 0.5|
+---+--------------------+---+----+
only showing top 5 rows

val dfExpRand = dfRand.withColumn("expo_sample", -($"one" / $"rate") * log($"one" - $"rand")) // samples from expo(rate=0.5)

dfExpRand: org.apache.spark.sql.DataFrame = [Id: bigint, rand: double ... 3 more fields]

dfExpRand.show(5)

+---+--------------------+---+----+--------------------+
| Id|                rand|one|rate|         expo_sample|
+---+--------------------+---+----+--------------------+
|  0|0.024042816995877625|1.0| 0.5|0.048673126808362034|
|  1|  0.4763832311243641|1.0| 0.5|  1.2939904386814756|
|  2|  0.3392911406256911|1.0| 0.5|  0.8288839819270684|
|  3|  0.6282132043405342|1.0| 0.5|  1.9788694379168241|
|  4|  0.9665960987271114|1.0| 0.5|   6.798165162486205|
+---+--------------------+---+----+--------------------+
only showing top 5 rows

display(dfExpRand)

dfExpRand.describe().show() // look sensible

+-------+-----------------+--------------------+----+----+--------------------+
|summary|               Id|                rand| one|rate|         expo_sample|
+-------+-----------------+--------------------+----+----+--------------------+
|  count|             1000|                1000|1000|1000|                1000|
|   mean|            499.5|  0.5060789944940257| 1.0| 0.5|   2.024647367046763|
| stddev|288.8194360957494| 0.28795893414434537| 0.0| 0.0|  1.9649069025265538|
|    min|                0|6.282087728304298E-4| 1.0| 0.5|0.001256812357281...|
|    max|              999|  0.9981606891187609| 1.0| 0.5|  12.596728597154938|
+-------+-----------------+--------------------+----+----+--------------------+

val expoSamplesDF = spark.range(1000000000).toDF("Id") // just make a DF of 100 row indices
               .select($"Id", rand(seed=1234567) as "rand") // add a column of random numbers in (0,1)
               .withColumn("one",lit(1.0))
               .withColumn("rate",lit(0.5))
               .withColumn("expo_sample", -($"one" / $"rate") * log($"one" - $"rand"))

expoSamplesDF: org.apache.spark.sql.DataFrame = [Id: bigint, rand: double ... 3 more fields]

expoSamplesDF.describe().show()

+-------+--------------------+--------------------+----------+----------+--------------------+
|summary|                  Id|                rand|       one|      rate|         expo_sample|
+-------+--------------------+--------------------+----------+----------+--------------------+
|  count|          1000000000|          1000000000|1000000000|1000000000|          1000000000|
|   mean|4.9999999907595944E8| 0.49999521358240573|       1.0|       0.5|  1.9999814119383708|
| stddev|2.8867513473915565E8|  0.2886816216562552|       0.0|       0.0|  2.0000011916404206|
|    min|                   0|1.584746778249268...|       1.0|       0.5|3.169493556749679...|
|    max|           999999999|  0.9999999996809935|       1.0|       0.5|  43.731619350261035|
+-------+--------------------+--------------------+----------+----------+--------------------+

Approximating Pi with Monte Carlo simulations

Uisng RDDs directly, let's estimate Pi.

//Calculate pi with Monte Carlo estimation
import scala.math.random

//make a very large unique set of 1 -> n 
val partitions = 2 
val n = math.min(100000L * partitions, Int.MaxValue).toInt 
val xs = 1 until n 

//split up n into the number of partitions we can use 
val rdd = sc.parallelize(xs, partitions).setName("'N values rdd'")

//generate a random set of points within a 2x2 square
val sample = rdd.map { i =>
  val x = random * 2 - 1
  val y = random * 2 - 1
  (x, y)
}.setName("'Random points rdd'")

//points w/in the square also w/in the center circle of r=1
val inside = sample.filter { case (x, y) => (x * x + y * y < 1) }.setName("'Random points inside circle'")
val count = inside.count()
 
//Area(circle)/Area(square) = inside/n => pi=4*inside/n                        
println("Pi is roughly " + 4.0 * count / n)

Pi is roughly 3.14156
import scala.math.random
partitions: Int = 2
n: Int = 200000
xs: scala.collection.immutable.Range = Range 1 until 200000
rdd: org.apache.spark.rdd.RDD[Int] = 'N values rdd' ParallelCollectionRDD[39] at parallelize at command-2971213210276568:10
sample: org.apache.spark.rdd.RDD[(Double, Double)] = 'Random points rdd' MapPartitionsRDD[40] at map at command-2971213210276568:13
inside: org.apache.spark.rdd.RDD[(Double, Double)] = 'Random points inside circle' MapPartitionsRDD[41] at filter at command-2971213210276568:20
count: Long = 157078

Doing it in PySpark is just as easy. This may be needed if there are pyhton libraries you want to take advantage of in Spark.

# # Estimating $\pi$
#
# This PySpark example shows you how to estimate $\pi$ in parallel
# using Monte Carlo integration.

from __future__ import print_function
import sys
from random import random
from operator import add

partitions = 2
n = 100000 * partitions

def f(_):
    x = random() * 2 - 1
    y = random() * 2 - 1
    return 1 if x ** 2 + y ** 2 < 1 else 0

# To access the associated SparkContext
count = spark.sparkContext.parallelize(range(1, n + 1), partitions).map(f).reduce(add)
print("Pi is roughly %f" % (4.0 * count / n))

Pi is roughly 3.138820

The following is from this google turotial.

Programming a Monte Carlo simulation in Scala

Monte Carlo, of course, is famous as a gambling destination. In this section, you use Scala to create a simulation that models the mathematical advantage that a casino enjoys in a game of chance. The "house edge" at a real casino varies widely from game to game; it can be over 20% in keno, for example. This tutorial creates a simple game where the house has only a one-percent advantage. Here's how the game works:

1. The player places a bet, consisting of a number of chips from a bankroll fund.

• The player rolls a 100-sided die (how cool would that be?).
• If the result of the roll is a number from 1 to 49, the player wins.
• For results 50 to 100, the player loses the bet.

You can see that this game creates a one-percent disadvantage for the player: in 51 of the 100 possible outcomes for each roll, the player loses.

Follow these steps to create and run the game:

val STARTING_FUND = 10
val STAKE = 1   // the amount of the bet
val NUMBER_OF_GAMES = 25

def rollDie: Int = {
    val r = scala.util.Random
    r.nextInt(99) + 1
}

def playGame(stake: Int): (Int) = {
    val faceValue = rollDie
    if (faceValue < 50)
        (2*stake)
    else
        (0)
}

// Function to play the game multiple times
// Returns the final fund amount
def playSession(
   startingFund: Int = STARTING_FUND,
   stake: Int = STAKE,
   numberOfGames: Int = NUMBER_OF_GAMES):
   (Int) = {

    // Initialize values
    var (currentFund, currentStake, currentGame) = (startingFund, 0, 1)

    // Keep playing until number of games is reached or funds run out
    while (currentGame <= numberOfGames && currentFund > 0) {

        // Set the current bet and deduct it from the fund
        currentStake = math.min(stake, currentFund)
        currentFund -= currentStake

        // Play the game
        val (winnings) = playGame(currentStake)

        // Add any winnings
        currentFund += winnings

        // Increment the loop counter
        currentGame += 1
    }
    (currentFund)
}

STARTING_FUND: Int = 10
STAKE: Int = 1
NUMBER_OF_GAMES: Int = 25
rollDie: Int
playGame: (stake: Int)Int
playSession: (startingFund: Int, stake: Int, numberOfGames: Int)Int

Enter the following code to play the game 25 times, which is the default value for NUMBER_OF_GAMES.

playSession()

res31: Int = 5

Your bankroll started with a value of 10 units. Is it higher or lower, now?

Now simulate 10,000 players betting 100 chips per game. Play 10,000 games in a session. This Monte Carlo simulation calculates the probability of losing all your money before the end of the session. Enter the follow code:

(sc.parallelize(1 to 10000, 500)
  .map(i => playSession(100000, 100, 250000))
  .map(i => if (i == 0) 1 else 0)
  .reduce(_+_)/10000.0)

res32: Double = 0.9992

Note that the syntax .reduce(_+_) is shorthand in Scala for aggregating by using a summing function.

The preceding code performs the following steps:

1. Creates an RDD with the results of playing the session.

• Replaces bankrupt players' results with the number 1 and nonzero results with the number 0.
• Sums the count of bankrupt players.
• Divides the count by the number of players.

A typical result might be:

res32: Double = 0.9992

Which represents a near guarantee of losing all your money, even though the casino had only a one-percent advantage.

Project Ideas

Try to create a scalable simulation of interest to you.

Here are some mature projects:

• https://github.com/zishanfu/GeoSparkSim
• https://github.com/srbaird/mc-var-spark

See a more complete VaR modeling: - https://databricks.com/blog/2020/05/27/modernizing-risk-management-part-1-streaming-data-ingestion-rapid-model-development-and-monte-carlo-simulations-at-scale.html

ScaDaMaLe Course site and book

Introduction to Machine Learning

Some very useful resources we will weave around for Statistical Learning, Data Mining, Machine Learning:

• January 2014, Stanford University professors Trevor Hastie and Rob Tibshirani (authors of the legendary Elements of Statistical Learning textbook) taught an online course based on their newest textbook, An Introduction to Statistical Learning with Applications in R (ISLR).
• https://www.dataschool.io/15-hours-of-expert-machine-learning-videos/
• free PDF of the ISLR book: http://www-bcf.usc.edu/~gareth/ISL/
• A more theoretically sound book with interesting aplications is Elements of Statistical Learning by the Stanford Gang of 3 (Hastie, Tibshirani and Friedman):
  • free PDF of the 10th printing: http://statweb.stanford.edu/~tibs/ElemStatLearn/printings/ESLII_print10.pdf
  • Solutions: http://waxworksmath.com/Authors/GM/Hastie/WriteUp/weatherwaxepsteinhastiesolutions_manual.pdf
• A great series on Probabilistic ML by Kevin P. Murphy https://probml.github.io/pml-book/.

Deep Learning is a very popular method currently (2017) in Machine Learning and I recommend Andrew Ng's free course in Courseera for this if you have not taken a course in deep learning already.

Note: This is an applied course in data science and we will quickly move to doing things with data. You have to do work to get a deeper mathematical understanding or take other courses. We will focus on intution here and the distributed ML Pipeline in action.

I am assuming several of you have or are taking ML courses. In this course we will focus on getting our hads dirty quickly with some level of common understanding across all the disciplines represented here. Please dive deep at your own time to desired depths and tangents based on your background.

Summary of Machine Learning at a High Level

• A rough definition of machine learning.
  • constructing and studying algorithms that learn from and make predictions on data.
• This broad area involves tools and ideas from various domains, including:
  • computer science,
  • probability and statistics,
  • optimization,
  • linear algebra
  • logic
  • etc.
• Common examples of ML, include:
  • facial recognition,
  • link prediction,
  • text or document classification, eg.::
    • spam detection,
  • protein structure prediction
  • teaching computers to play games (go!)

Some common terminology

using example of spam detection as a running example.

• the data points we learn from are call observations:

  • they are items or entities used for::
    • learning or
    • evaluation.

• in the context of spam detection,

  • emails are our observations.
  • Features are attributes used to represent an observation.
  • Features are typically numeric,
    • and in spam detection, they can be:
    • the length,
    • the date, or
    • the presence or absence of keywords in emails.
  • Labels are values or categories assigned to observations.
    • and in spam detection, they can be:
    • an email being defined as spam or not spam.

• Training and test data sets are the observations that we use to train and evaluate a learning algorithm.

• Pop-Quiz

  • What is the difference between supervised and unsupervised learning?

For a Stats@Stanford Hastie-Tibshirani Perspective on Supervised and Unsupervised Learning:

(watch later 12:12):

From : https://www.dataschool.io/15-hours-of-expert-machine-learning-videos/.

ML Pipelines

Here we will use ML Pipelines to do machine learning at scale.

See https://spark.apache.org/docs/latest/ml-pipeline.html for a quick overview.

Read this section for an overview:

• https://learning.oreilly.com/library/view/spark-the-definitive/9781491912201/ch24.html#high-level-mllib-concepts.

ScaDaMaLe Course site and book

Million Song Dataset - Kaggle Challenge

Predict which songs a user will listen to.

SOURCE: This is just a Scala-rification of the Python notebook published in databricks community edition in 2016.

When you first hear a song, do you ever categorize it as slow or fast in your head? Is it even a valid categorization? If so, can one do it automatically? I have always wondered about that. That is why I got excited when I learned about the Million Songs Dataset -Kaggle Challenge.

In this tutorial we will walk through a practical example of a data science project with Apache Spark in Python. We are going to parse, explore and model a sample from the million songs dataset stored on distributed storage. This tutorial is organized into three sections:

1. ETL: Parses raw texts and creates a cached table
2. Explore: Explores different aspects of the songs table using graphs
3. Model: Uses SparkML to cluster songs based on some of their attributes

The goal of this tutorial is to prepare you for real world data science projects. Make sure you go through the tutorial in the above order and use the exercises to make yourself familiar further with the API. Also make sure you run these notebooks on a 1.6.x cluster.

1. ETL

The first step of most data science projects is extracting, transforming and loading data into well formated tables. Our example starts with ETL as well. By following the ETL noteboook you can expect to learn about following Spark concepts:

• RDD: Resilient Distributed Dataset
• Reading and transforming RDDs
• Schema in Spark
• Spark DataFrame
• Temp tables
• Caching tables

2. Explore

Exploratory analysis is a key step in any real data project. Data scientists use variety of tools to explore and visualize their data. In the second notebook of this tutorial we introduce several tools in Python and Databricks notebooks that can help you visually explore your large data. By reading this notebook and finishing its exercises you will become familiar with:

• How to view the schema of a table
• How to display ggplot and matplotlib figures in Notebooks
• How to summarize and visualize different aspects of large datasets
• How to sample and visualize large data

3. Model

The end goal of many data scientists is producing useful models. These models are often used for prediction of new and upcoming events in production. In our third notebook we construct a simple K-means clustering model. In this notebook you will learn about:

• Feature transformation
• Fitting a model using SparkML API
• Applying a model to data
• Visualizing model results
• Model tuning

ScaDaMaLe Course site and book

Million Song Dataset - Kaggle Challenge

Predict which songs a user will listen to.

SOURCE: This is just a Scala-rification of the Python notebook published in databricks community edition in 2016.

Stage 2: Exploring songs data

This is the second notebook in this tutorial. In this notebook we do what any data scientist does with their data right after parsing it: exploring and understanding different aspect of data. Make sure you understand how we get the songsTable by reading and running the ETL notebook. In the ETL notebook we created and cached a temporary table named songsTable

Let's Do all the main bits in Stage 1 now before doing Stage 2 in this Notebook.

// Let's quickly do everything to register the tempView of the table here

// fill in comment ... EXERCISE!
case class Song(artist_id: String, artist_latitude: Double, artist_longitude: Double, artist_location: String, artist_name: String, duration: Double, end_of_fade_in: Double, key: Int, key_confidence: Double, loudness: Double, release: String, song_hotness: Double, song_id: String, start_of_fade_out: Double, tempo: Double, time_signature: Double, time_signature_confidence: Double, title: String, year: Double, partial_sequence: Int)

def parseLine(line: String): Song = {
  // fill in comment ...
  
  def toDouble(value: String, defaultVal: Double): Double = {
    try {
       value.toDouble
    } catch {
      case e: Exception => defaultVal
    }
  }

  def toInt(value: String, defaultVal: Int): Int = {
    try {
       value.toInt
      } catch {
      case e: Exception => defaultVal
    }
  }
  // fill in comment ...
  val tokens = line.split("\t")
  Song(tokens(0), toDouble(tokens(1), 0.0), toDouble(tokens(2), 0.0), tokens(3), tokens(4), toDouble(tokens(5), 0.0), toDouble(tokens(6), 0.0), toInt(tokens(7), -1), toDouble(tokens(8), 0.0), toDouble(tokens(9), 0.0), tokens(10), toDouble(tokens(11), 0.0), tokens(12), toDouble(tokens(13), 0.0), toDouble(tokens(14), 0.0), toDouble(tokens(15), 0.0), toDouble(tokens(16), 0.0), tokens(17), toDouble(tokens(18), 0.0), toInt(tokens(19), -1))
}

// this is loads all the data - a subset of the 1M songs dataset
val dataRDD = sc.textFile("/datasets/sds/songs/data-001/part-*") 

// .. fill in comment
val df = dataRDD.map(parseLine).toDF

// .. fill in comment
df.createOrReplaceTempView("songsTable")

defined class Song
parseLine: (line: String)Song
dataRDD: org.apache.spark.rdd.RDD[String] = /databricks-datasets/songs/data-001/part-* MapPartitionsRDD[81] at textFile at command-2971213210276755:30
df: org.apache.spark.sql.DataFrame = [artist_id: string, artist_latitude: double ... 18 more fields]

spark.catalog.listTables.show(false) // make sure the temp view of our table is there

+----------------------------+--------+-----------+---------+-----------+
|name                        |database|description|tableType|isTemporary|
+----------------------------+--------+-----------+---------+-----------+
|countrycodes                |default |null       |EXTERNAL |false      |
|ltcar_locations_2_csv       |default |null       |MANAGED  |false      |
|magellan                    |default |null       |MANAGED  |false      |
|mobile_sample               |default |null       |EXTERNAL |false      |
|over300all_2_txt            |default |null       |MANAGED  |false      |
|person                      |default |null       |MANAGED  |false      |
|persons                     |default |null       |MANAGED  |false      |
|social_media_usage          |default |null       |MANAGED  |false      |
|social_media_usage_csv_gui  |default |null       |MANAGED  |false      |
|voronoi20191213uppsla1st_txt|default |null       |MANAGED  |false      |
|voronoi20191213uppsla2d_txt |default |null       |MANAGED  |false      |
|voronoi20191213uppsla3d_txt |default |null       |MANAGED  |false      |
|songstable                  |null    |null       |TEMPORARY|true       |
+----------------------------+--------+-----------+---------+-----------+

A first inspection

A first step to any data exploration is viewing sample data. For this purpose we can use a simple SQL query that returns first 10 rows.

select * from songsTable limit 10

table("songsTable").printSchema()

root
 |-- artist_id: string (nullable = true)
 |-- artist_latitude: double (nullable = false)
 |-- artist_longitude: double (nullable = false)
 |-- artist_location: string (nullable = true)
 |-- artist_name: string (nullable = true)
 |-- duration: double (nullable = false)
 |-- end_of_fade_in: double (nullable = false)
 |-- key: integer (nullable = false)
 |-- key_confidence: double (nullable = false)
 |-- loudness: double (nullable = false)
 |-- release: string (nullable = true)
 |-- song_hotness: double (nullable = false)
 |-- song_id: string (nullable = true)
 |-- start_of_fade_out: double (nullable = false)
 |-- tempo: double (nullable = false)
 |-- time_signature: double (nullable = false)
 |-- time_signature_confidence: double (nullable = false)
 |-- title: string (nullable = true)
 |-- year: double (nullable = false)
 |-- partial_sequence: integer (nullable = false)

select count(*) from songsTable

table("songsTable").count() // or equivalently with DataFrame API - recall table("songsTable") is a DataFrame

res4: Long = 31369

display(sqlContext.sql("SELECT duration, year FROM songsTable")) // Aggregation is set to 'Average' in 'Plot Options'